package spring.cloud.tasks.api.service.impl;

import com.google.common.base.Strings;
import lombok.Getter;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.curator.framework.CuratorFramework;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.util.CollectionUtils;
import spring.cloud.tasks.api.database.entity.NameSpaceDto;
import spring.cloud.tasks.api.database.entity.ZooKeeperClusterDto;
import spring.cloud.tasks.api.database.repository.NameSpaceDtoRepository;
import spring.cloud.tasks.api.database.repository.NameSpaceZooKeeperClusterMappingDtoRepository;
import spring.cloud.tasks.api.database.repository.TaskConfigurationDtoRepository;
import spring.cloud.tasks.api.database.service.NameSpaceDtoService;
import spring.cloud.tasks.api.database.service.NameSpaceZooKeeperClusterMappingDtoService;
import spring.cloud.tasks.api.database.service.ZooKeeperClusterDtoService;
import spring.cloud.tasks.api.domain.executor.ExecutorInfo;
import spring.cloud.tasks.api.domain.task_executor.TaskExecutorStatus;
import spring.cloud.tasks.api.domain.zooKeeperClusterAndNameSpace.NameSpace;
import spring.cloud.tasks.api.domain.zooKeeperClusterAndNameSpace.NameSpaceConfiguration;
import spring.cloud.tasks.api.domain.zooKeeperClusterAndNameSpace.NameSpaceZooKeeperClient;
import spring.cloud.tasks.api.domain.zooKeeperClusterAndNameSpace.ZooKeeperCluster;
import spring.cloud.tasks.api.service.ExecutorService;
import spring.cloud.tasks.api.service.ZooKeeperClusterService;
import spring.cloud.tasks.common.node.ExecutorNodePath;
import spring.cloud.tasks.common.node.TaskNodePath;
import spring.cloud.tasks.common.zookeeper.ZooKeeper;
import spring.cloud.tasks.common.zookeeper.ZooKeeperRepository;

import javax.annotation.Resource;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;

@Setter
@Getter
@Slf4j
public class ZooKeeperClusterServiceImpl implements ZooKeeperClusterService {
    @Resource
    private ZooKeeperClusterDtoService zooKeeperClusterDtoService;
    @Resource
    private NameSpaceZooKeeperClusterMappingDtoService nameSpaceZooKeeperClusterMappingDtoService;
    @Resource
    private NameSpaceDtoService nameSpaceDtoService;
    @Resource
    private ZooKeeperRepository zooKeeperRepository;
    @Autowired
    private ExecutorService executorService;

    @Autowired
    private NameSpaceDtoRepository nameSpaceDtoRepository;

    @Autowired
    private TaskConfigurationDtoRepository taskConfigurationDtoRepository;

    @Autowired
    private NameSpaceZooKeeperClusterMappingDtoRepository nameSpaceZooKeeperClusterMappingDtoRepository;

    @Autowired
    private ZooKeeperClusterService zooKeeperClusterService;


    private final ConcurrentHashMap<String, Object> nameSpaceLock = new ConcurrentHashMap<>(); // maybe could remove in right time
    private final ConcurrentHashMap<String, NameSpaceZooKeeperClient> nameSpaceToNameSpaceZooKeeperClientMap = new ConcurrentHashMap<>();
    //


    /**
     * 为保证values有序
     **/
    private final LinkedHashMap<String, ZooKeeperCluster> zooKeeperClusterKeyToZooKeeperClusterMap = new LinkedHashMap<>();
    private final List<String> onlineNameSpaceList = new ArrayList<>();


    @Override
    public void notifyRegistryCenterRefresh() throws RuntimeException {
        //DB更新 UUID
    }

    @Transactional(rollbackFor = Exception.class)
    @Override
    public void addZooKeeperCluster(String zooKeeperClusterKey, String zooKeeperAddressList, String digest, String description) throws RuntimeException {
        ZooKeeperClusterDto zooKeeperClusterDto = zooKeeperClusterDtoService.selectByZooKeeperClusterKey(zooKeeperClusterKey);
        if (zooKeeperClusterDto != null) {
            throw new RuntimeException(String.format("ZooKeeperCluster[%s]已经存在", zooKeeperClusterKey));
        }
        zooKeeperClusterDtoService.insert(zooKeeperClusterKey, zooKeeperAddressList, digest, description, "");
        ZooKeeperCluster zooKeeperCluster = new ZooKeeperCluster(zooKeeperClusterKey, zooKeeperAddressList, digest, description);
        zooKeeperClusterKeyToZooKeeperClusterMap.put(zooKeeperClusterKey, zooKeeperCluster);
        notifyRegistryCenterRefresh();
    }

    @Transactional(rollbackFor = Exception.class)
    @Override
    public void updateZooKeeperCluster(String zooKeeperClusterKey, String connectString, String description) throws RuntimeException {
        ZooKeeperClusterDto zooKeeperClusterDto = zooKeeperClusterDtoService.selectByZooKeeperClusterKey(zooKeeperClusterKey);
        if (zooKeeperClusterDto == null) {
            throw new RuntimeException(String.format("ZooKeeperCluster[%s]不存在", zooKeeperClusterKey));
        }
        zooKeeperClusterDto.setZooKeeperClusterAddressList(connectString);
        zooKeeperClusterDto.setDescription(description);
        zooKeeperClusterDto.setLastUpdateTime(new Date());
        zooKeeperClusterDtoService.updateByZooKeeperClusterKey(zooKeeperClusterDto);
        notifyRegistryCenterRefresh();
    }

    @Transactional(rollbackFor = Exception.class)
    @Override
    public void deleteZooKeeperCluster(String zooKeeperClusterKey) throws RuntimeException {
        if (getZooKeeperCluster(zooKeeperClusterKey) == null) {
            throw new RuntimeException("ZooKeeperCluster不存在");
        }
        if (nameSpaceCount(zooKeeperClusterKey) > 0) {
            throw new RuntimeException("ZooKeeperCluster下NameSpace不为空");
        }
        if (!CollectionUtils.isEmpty(nameSpaceZooKeeperClusterMappingDtoService.getNameSpaceList(zooKeeperClusterKey))) {
            throw new RuntimeException("ZooKeeperCluster下NameSpace不为空");
        }
        zooKeeperClusterDtoService.deleteByZooKeeperClusterKey(zooKeeperClusterKey);
        zooKeeperClusterKeyToZooKeeperClusterMap.remove(zooKeeperClusterKey);
        notifyRegistryCenterRefresh();
    }


    @Override
    public List<ZooKeeperCluster> getZooKeeperClusterList() {
        return new ArrayList<>(zooKeeperClusterKeyToZooKeeperClusterMap.values());
    }

    @Override
    public List<ZooKeeperCluster> getOnlineZooKeeperClusterList() {
        List<ZooKeeperCluster> zooKeeperClusterList = new ArrayList<>();
        for (ZooKeeperCluster zooKeeperCluster : zooKeeperClusterKeyToZooKeeperClusterMap.values()) {
            if (!zooKeeperCluster.isOffline()) {
                zooKeeperClusterList.add(zooKeeperCluster);
            }
        }
        return zooKeeperClusterList;
    }

    @Override
    public ZooKeeperCluster getZooKeeperCluster(String zooKeeperClusterKey) {
        return zooKeeperClusterKeyToZooKeeperClusterMap.get(zooKeeperClusterKey);
    }

    @Override
    public List<String> getOnlineNameSpaceList() throws RuntimeException {
        return onlineNameSpaceList;
    }

    @Override
    public List<NameSpaceConfiguration> getNameSpaceConfigurationList(ZooKeeperCluster zooKeeperCluster) {
        List<String> nameSpaceList = nameSpaceZooKeeperClusterMappingDtoService.getNameSpaceList(zooKeeperCluster.getZooKeeperClusterKey());
        if (nameSpaceList == null || nameSpaceList.isEmpty()) {
            return new ArrayList<>();
        }

        List<NameSpaceConfiguration> nameSpaceConfigurationList = new ArrayList<>();
        for (String nameSpace : nameSpaceList) {
            NameSpaceConfiguration nameSpaceConfiguration = constructNameSpaceConfiguration(zooKeeperCluster, nameSpace);
            nameSpaceConfigurationList.add(nameSpaceConfiguration);
        }

        return nameSpaceConfigurationList;
    }

    protected NameSpaceConfiguration constructNameSpaceConfiguration(ZooKeeperCluster zooKeeperCluster, String nameSpace) {
        NameSpaceConfiguration nameSpaceConfiguration = new NameSpaceConfiguration();
        nameSpaceConfiguration.setZooKeeperClusterKey(zooKeeperCluster.getZooKeeperClusterKey());
        nameSpaceConfiguration.setZooKeeperAddressList(zooKeeperCluster.getZooKeeperAddressList());
        nameSpaceConfiguration.setNameSpace(nameSpace);

        return nameSpaceConfiguration;
    }

    @Override
    public int nameSpaceCount(String zooKeeperClusterKey) {
        ZooKeeperCluster zooKeeperCluster = zooKeeperClusterKeyToZooKeeperClusterMap.get(zooKeeperClusterKey);
        if (zooKeeperCluster != null) {
            List<NameSpaceConfiguration> nameSpaceConfigurationList = zooKeeperCluster.getNameSpaceConfigurationList();
            if (nameSpaceConfigurationList != null) {
                return nameSpaceConfigurationList.size();
            }
        }
        return 0;
    }

    @Override
    public NameSpaceConfiguration getNameSpaceConfiguration(String nameSpace) {
        if (Strings.isNullOrEmpty(nameSpace)) {
            return null;
        }
        Collection<ZooKeeperCluster> collection = zooKeeperClusterKeyToZooKeeperClusterMap.values();
        for (ZooKeeperCluster zooKeeperCluster : collection) {
            for (NameSpaceConfiguration nameSpaceConfiguration : zooKeeperCluster.getNameSpaceConfigurationList()) {
                if (nameSpaceConfiguration != null && nameSpace.equals(nameSpaceConfiguration.getNameSpace())) {
                    return nameSpaceConfiguration;
                }
            }
        }
        return null;
    }

    @Override
    public NameSpace getNameSpace(String nameSpace) throws RuntimeException {
        if (nameSpaceDtoService.selectByNameSpaceKey(nameSpace) == null) {
            throw new RuntimeException("NameSpace" + nameSpace + "不存在");
        }

        String zooKeeperClusterKey = nameSpaceZooKeeperClusterMappingDtoService.getZooKeeperClusterKey(nameSpace);
        if (StringUtils.isBlank(zooKeeperClusterKey)) {
            throw new RuntimeException("ZooKeeperCluster" + zooKeeperClusterKey + "不存在");
        }

        NameSpace nameSpace1 = new NameSpace();
        nameSpace1.setZooKeeperClusterKey(zooKeeperClusterKey);
        nameSpace1.setNameSpace(nameSpace);
        return nameSpace1;
    }


    //##<!-我是优美的功能模块分割线->##


    @Transactional(rollbackFor = {Exception.class})
    @Override
    public void createNameSpace(NameSpace nameSpace, String userId) throws RuntimeException {
        String zooKeeperClusterKey = nameSpace.getZooKeeperClusterKey();
        String nameSpaceString = nameSpace.getNameSpace();
        ZooKeeperCluster zooKeeperCluster = getZooKeeperCluster(zooKeeperClusterKey);
        if (zooKeeperCluster == null) {
            throw new RuntimeException("ZooKeeperCluster" + zooKeeperClusterKey + "不存在");
        }
        if (nameSpaceDtoService.selectByNameSpaceKey(nameSpaceString) != null) {
            throw new RuntimeException("NameSpace" + nameSpace + "已经存在");
        }

        try {
            // 创建 nameSpace
            NameSpaceDto nameSpaceDto = constructNameSpaceDto(nameSpace, userId);
            nameSpaceDtoService.insert(nameSpaceDto);
            // 创建 NameSpace 和 ZooKeeperCluster 映射关系
            nameSpaceZooKeeperClusterMappingDtoService.insert(zooKeeperClusterKey, nameSpaceString);
            NameSpaceConfiguration nameSpaceConfiguration = new NameSpaceConfiguration();
            nameSpaceConfiguration.setZooKeeperClusterKey(zooKeeperClusterKey);
            nameSpaceConfiguration.setNameSpace(nameSpaceString);
            zooKeeperCluster.getNameSpaceConfigurationList().add(nameSpaceConfiguration);
            // refresh
            notifyRegistryCenterRefresh();
        } catch (Exception e) {
            log.error(e.getMessage(), e);
            throw new RuntimeException("创建NameSpace:" + nameSpace + "失败", e);
        }
    }

    private NameSpaceDto constructNameSpaceDto(NameSpace nameSpace, String userId) {
        NameSpaceDto nameSpaceDto = new NameSpaceDto();
        nameSpaceDto.setCreatedBy(userId);
        nameSpaceDto.setCreateTime(new Date());
        nameSpaceDto.setLastUpdatedBy(userId);
        nameSpaceDto.setLastUpdateTime(new Date());
        //
        nameSpaceDto.setNameSpaceKey(nameSpace.getNameSpace());
        nameSpaceDto.setDescription(nameSpace.getContent());
        nameSpaceDto.setIsEffective(1);

        return nameSpaceDto;
    }

    @Override
    public void deleteNameSpace(String nameSpace) throws RuntimeException {
        boolean online = isExecutorsOnline(nameSpace);
        if (online) {
            log.info("nameSpace {} has online executor, can not delete it", nameSpace);
            throw new RuntimeException("nameSpace has online executor, can not delete it");
        } else {
            NameSpaceConfiguration nameSpaceConfiguration = zooKeeperClusterService.getNameSpaceConfiguration(nameSpace);
            deleteNameSpaceInZooKeeper(nameSpaceConfiguration, nameSpace);
            deleteInfosInDB(nameSpace);
        }
    }


    protected boolean isExecutorsOnline(String nameSpace) throws RuntimeException {
        List<ExecutorInfo> executorInfoList = executorService.getExecutorInfoList(nameSpace, TaskExecutorStatus.ONLINE);
        return executorInfoList.size() > 0;
    }

    @Transactional(rollbackFor = Exception.class)
    protected void deleteInfosInDB(String nameSpace) {
        deleteNameSpaceInDB(nameSpace);
        log.info("delete nameSpace in DB success - nameSpace {}", nameSpace);
        deleteNameSpaceZooKeeperClusterMappingInDb(nameSpace);
        log.info("delete nameSpaceZkCluster in DB success - nameSpace {}", nameSpace);
        deleteTasksInDB(nameSpace);
        log.info("delete jobs in DB success - nameSpace {}", nameSpace);
    }

    protected void deleteTasksInDB(String nameSpace) {
        taskConfigurationDtoRepository.deleteByNameSpace(nameSpace);
    }

    protected void deleteNameSpaceZooKeeperClusterMappingInDb(String nameSpace) {
        nameSpaceZooKeeperClusterMappingDtoRepository.deleteByNameSpaceKey(nameSpace);
    }

    protected void deleteNameSpaceInDB(String nameSpace) {
        nameSpaceDtoRepository.deleteByNameSpace(nameSpace);
    }


    protected void deleteNameSpaceInZooKeeper(NameSpaceConfiguration nameSpaceConfiguration, String nameSpace) {
        CuratorFramework curatorFramework = null;
        try {
            curatorFramework = zooKeeperRepository.connect(nameSpaceConfiguration.getZooKeeperAddressList(), nameSpaceConfiguration.getDigest(), null
            );
            curatorFramework.delete().deletingChildrenIfNeeded().forPath("/" + nameSpace);
            log.info("delete nameSpace in zk success - nameSpace {}", nameSpace);
        } catch (Exception e) {
            log.warn("fail to delete nameSpace:{}", nameSpace, e);
        } finally {
            if (curatorFramework != null) {
                curatorFramework.close();
            }
        }
    }

    /**
     * 核心方法
     */
    @Override
    public ZooKeeper connectZooKeeper(String connectString, String nameSpace) throws RuntimeException {
        CuratorFramework curatorFramework = zooKeeperRepository.connect(connectString, null, nameSpace);
        if (curatorFramework != null) {
            return zooKeeperRepository.getZooKeeper(curatorFramework);
        }
        return null;
    }

    @Override
    public boolean nameSpaceCheck(CuratorFramework curatorFramework, String nameSpace) throws RuntimeException {
        try {
            String tasksPath = "/" + nameSpace + TaskNodePath.getTasksNodePath();
            if (!(curatorFramework.checkExists().forPath(tasksPath) != null)) {
                return false;
            }
            String executorsPath = "/" + nameSpace + ExecutorNodePath.getExecutorsNodePath();
            return curatorFramework.checkExists().forPath(executorsPath) != null;
        } catch (Exception e) {
            log.error(e.getMessage(), e);
            return false;
        }
    }

    /**
     * 核心方法
     */
    @Override
    public NameSpaceZooKeeperClient getNameSpaceZooKeeperClient(String nameSpace) {
        NameSpaceConfiguration nameSpaceConfiguration = getNameSpaceConfiguration(nameSpace);
        if (nameSpaceConfiguration == null) {
            return null;
        }
        String zooKeeperAddressList = nameSpaceConfiguration.getZooKeeperAddressList();
        String digest = nameSpaceConfiguration.getDigest();
        synchronized (getNameSpaceLock(nameSpace)) {
            NameSpaceZooKeeperClient nameSpaceZooKeeperClient = nameSpaceToNameSpaceZooKeeperClientMap.get(nameSpace);
            if (nameSpaceZooKeeperClient != null) {
                CuratorFramework curatorFramework = nameSpaceZooKeeperClient.getCuratorFramework();
                if (curatorFramework != null && curatorFramework.getZookeeperClient().isConnected()) {
                    nameSpaceZooKeeperClient.setConnected(true);
                    return nameSpaceZooKeeperClient;
                } else {
                    nameSpaceToNameSpaceZooKeeperClientMap.remove(nameSpace);
                    nameSpaceZooKeeperClient.setConnected(false);
                    nameSpaceZooKeeperClient.setCuratorFramework(null);
                }
            }
            nameSpaceZooKeeperClient = new NameSpaceZooKeeperClient();
            nameSpaceZooKeeperClient.setNameSpace(nameSpace);
            nameSpaceZooKeeperClient.setZooKeeperAddressList(zooKeeperAddressList);
            CuratorFramework curatorFramework = zooKeeperRepository.connect(zooKeeperAddressList, digest, nameSpace);
            if (curatorFramework != null && curatorFramework.getZookeeperClient().isConnected()) {
                nameSpaceZooKeeperClient.setConnected(true);
                nameSpaceZooKeeperClient.setCuratorFramework(curatorFramework);
                nameSpaceToNameSpaceZooKeeperClientMap.put(nameSpace, nameSpaceZooKeeperClient);
                return nameSpaceZooKeeperClient;
            }
        }
        return null;
    }

    /**
     * 核心方法
     */
    @Override
    public ZooKeeper connectZooKeeper(String nameSpace) throws RuntimeException {
        ZooKeeper zooKeeper = null;
        try {
            NameSpaceConfiguration nameSpaceConfiguration = getNameSpaceConfiguration(nameSpace);
            if (nameSpaceConfiguration == null) {
                throw new RuntimeException("find registryCenterConfiguration failed");
            }
            String zooKeeperAddressList = nameSpaceConfiguration.getZooKeeperAddressList();
            String digest = nameSpaceConfiguration.getDigest();
            synchronized (getNameSpaceLock(nameSpace)) {
                NameSpaceZooKeeperClient nameSpaceZooKeeperClient = nameSpaceToNameSpaceZooKeeperClientMap.get(nameSpace);
                if (nameSpaceZooKeeperClient != null) {
                    CuratorFramework curatorFramework = nameSpaceZooKeeperClient.getCuratorFramework();
                    if (curatorFramework != null && curatorFramework.getZookeeperClient().isConnected()) {
                        nameSpaceZooKeeperClient.setConnected(true);
                        zooKeeper = zooKeeperRepository.getZooKeeper(curatorFramework);
                        return zooKeeper;
                    } else {
                        nameSpaceToNameSpaceZooKeeperClientMap.remove(nameSpace);
                        nameSpaceZooKeeperClient.setConnected(false);
                        nameSpaceZooKeeperClient.setCuratorFramework(null);
                    }
                }
                nameSpaceZooKeeperClient = new NameSpaceZooKeeperClient();
                nameSpaceZooKeeperClient.setNameSpace(nameSpace);
                nameSpaceZooKeeperClient.setZooKeeperAddressList(zooKeeperAddressList);
                CuratorFramework curatorFramework = zooKeeperRepository.connect(zooKeeperAddressList, digest, nameSpace);
                if (curatorFramework != null && curatorFramework.getZookeeperClient().isConnected()) {
                    nameSpaceZooKeeperClient.setConnected(true);
                    nameSpaceZooKeeperClient.setCuratorFramework(curatorFramework);
                    nameSpaceToNameSpaceZooKeeperClientMap.put(nameSpace, nameSpaceZooKeeperClient);
                    zooKeeper = zooKeeperRepository.getZooKeeper(curatorFramework);
                    return zooKeeper;
                }
            }
        } catch (RuntimeException e) {
            throw e;
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
        if (zooKeeper == null) {
            throw new RuntimeException("connect zookeeper failed");
        }
        return zooKeeper;
    }

    private Object getNameSpaceLock(String nameSpace) {
        Object lock = nameSpaceLock.get(nameSpace);
        if (lock == null) {
            lock = new Object();
            Object pre = nameSpaceLock.putIfAbsent(nameSpace, lock);
            if (pre != null) {
                lock = pre;
            }
        }
        return lock;
    }


}
